package dmp.cjh.url.classfy;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class MyOutputFormat<K, V> extends FileOutputFormat<K, V> {

	@Override
	public RecordWriter<K, V> getRecordWriter(TaskAttemptContext arg0) throws IOException, InterruptedException {
		FileSystem fs = FileSystem.get(new Configuration());
		FSDataOutputStream qczj_search_url = fs.create(new Path("/user/vendorwins/chenjinghui/urlfilter/qczj_search_url"));
		FSDataOutputStream qczj_search_by_brand = fs.create(new Path("/user/vendorwins/chenjinghui/urlfilter/qczj_search_by_brand"));
		FSDataOutputStream qczj_search_consult_price = fs.create(new Path("/user/vendorwins/chenjinghui/urlfilter/qczj_search_consult_price"));
		FSDataOutputStream qcbjdq_search_by_brand = fs.create(new Path("/user/vendorwins/chenjinghui/urlfilter/qcbjdq_search_by_brand"));
		FSDataOutputStream qcbjdq_search_lowest_price = fs.create(new Path("/user/vendorwins/chenjinghui/urlfilter/qcbjdq_search_lowest_price"));
		FSDataOutputStream hmc_search_lowest_price = fs.create(new Path("/user/vendorwins/chenjinghui/urlfilter/hmc_search_lowest_price"));
		FSDataOutputStream hmc_app_search_lowest_price = fs.create(new Path("/user/vendorwins/chenjinghui/urlfilter/hmc_app_search_lowest_price"));

		return new MyRecordWriter<K, V>(qczj_search_url, qczj_search_by_brand, qczj_search_consult_price, qcbjdq_search_by_brand, qcbjdq_search_lowest_price, hmc_search_lowest_price, hmc_app_search_lowest_price);
	}

	public static class MyRecordWriter<K, V> extends RecordWriter<K, V> {
		private FSDataOutputStream qczj_search_url = null;
		private FSDataOutputStream qczj_search_by_brand = null;
		private FSDataOutputStream qczj_search_consult_price = null;
		private FSDataOutputStream qcbjdq_search_by_brand = null;
		private FSDataOutputStream qcbjdq_search_lowest_price = null;
		private FSDataOutputStream hmc_search_lowest_price = null;
		private FSDataOutputStream hmc_app_search_lowest_price = null;

		public MyRecordWriter() {

		}

		public MyRecordWriter(FSDataOutputStream qczj_search_url, FSDataOutputStream qczj_search_by_brand, FSDataOutputStream qczj_search_consult_price, FSDataOutputStream qcbjdq_search_by_brand, FSDataOutputStream qcbjdq_search_lowest_price, FSDataOutputStream hmc_search_lowest_price, FSDataOutputStream hmc_app_search_lowest_price) {
			super();
			this.qczj_search_url = qczj_search_url;
			this.qczj_search_by_brand = qczj_search_by_brand;
			this.qczj_search_consult_price = qczj_search_consult_price;
			this.qcbjdq_search_by_brand = qcbjdq_search_by_brand;
			this.qcbjdq_search_lowest_price = qcbjdq_search_lowest_price;
			this.hmc_search_lowest_price = hmc_search_lowest_price;
			this.hmc_app_search_lowest_price = hmc_app_search_lowest_price;
		}

		@Override
		public void close(TaskAttemptContext arg0) throws IOException, InterruptedException {

			if (qczj_search_url != null) {
				qczj_search_url.close();
			}
			if (qczj_search_by_brand != null) {
				qczj_search_by_brand.close();
			}
			if (qczj_search_consult_price != null) {
				qczj_search_consult_price.close();
			}
			if (qcbjdq_search_by_brand != null) {
				qcbjdq_search_by_brand.close();
			}
			if (qcbjdq_search_lowest_price != null) {
				qcbjdq_search_lowest_price.close();
			}
			if (hmc_search_lowest_price != null) {
				hmc_search_lowest_price.close();
			}
			if (hmc_app_search_lowest_price != null) {
				hmc_app_search_lowest_price.close();
			}

		}

		@Override
		public void write(K key, V value) throws IOException, InterruptedException {

			if (key.toString().contains("qczj_search_url")) {
				qczj_search_url.write((value.toString()+"\r\n").getBytes());
			}
			if (key.toString().contains("qczj_search_by_brand")) {
				qczj_search_by_brand.write((value.toString()+"\r\n").getBytes());
			}
			if (key.toString().contains("qczj_search_consult_price")) {
				qczj_search_consult_price.write((value.toString()+"\r\n").getBytes());
			}
			if (key.toString().contains("qcbjdq_search_by_brand")) {
				qcbjdq_search_by_brand.write((value.toString()+"\r\n").getBytes());
			}
			if (key.toString().contains("qcbjdq_search_lowest_price")) {
				qcbjdq_search_lowest_price.write((value.toString()+"\r\n").getBytes());
			}
			if (key.toString().contains("hmc_search_lowest_price")) {
				hmc_search_lowest_price.write((value.toString()+"\r\n").getBytes());
			}
			if (key.toString().contains("hmc_app_search_lowest_price")) {
				hmc_app_search_lowest_price.write((value.toString()+"\r\n").getBytes());
			}
		}

	}

}
